package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.2 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C20_Cogroup {

  def main(args: Array[String]): Unit = {

    val sc = SparkUtil.getSparkContext("cogroup")

    val arr1 = Array(("a", 11), ("c", 22), ("d", 33), ("d", 333), ("a", 111), ("b", 222), ("f", 2))
    val arr2 = Array(("a", 1111), ("c", 2222), ("d", 3333), ("o", 333), ("a", 111), ("b", 422), ("f", 21))
    val rdd1= sc.parallelize(arr1,2)
    val rdd2 = sc.parallelize(arr2 )

    /**
     * 对两个RDD中的数据 分别分组  rdd1-->迭代器   rdd2--->迭代器  迭代器中只存储value值
     * 组名的一致
     */
    val resRDD: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)

  /*  rdd1有两个分区 , 每个分区的数据
    rdd1.mapPartitionsWithIndex()*/

/*
    val res: RDD[Array[(String, Int)]] = rdd1.glom()
    res.map(arr=>{
      arr.size
    }).foreach(println)   //  3  4
*/


    val ls = List(1,2,3,4,5,6,7,8,9,10,11,12,14,12,16,23,17,43,32,78)

    val rdd = sc.makeRDD(ls, 3)
   // val rr = rdd.sortBy(e => e)
       // 每个分区的最大值
    val value: RDD[Array[Int]] = rdd.glom()
    // 展示每个分区的数据
    val res: Array[Array[Int]] = value.collect()
    res.foreach(arr=>println(arr.toList))


    sc.stop



  }

}
